package reactor;

import reactor.core.publisher.Flux;
import reactor.core.publisher.SynchronousSink;
import reactor.core.scheduler.Schedulers;
import util.Util;

import java.util.concurrent.CountDownLatch;
import java.util.function.BiFunction;

/**
 * @author yding5
 */
public class PublishOnTest {

    private static final CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * 状态感知型下发
     * 
     * @param args
     * @throws InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
        // 定义初始状态为0
        Flux.generate(() -> 0, (BiFunction<Integer, SynchronousSink<Integer>, Integer>)(state, sink) -> {
            if (state == 5) {
                // state更新至5时，结束生产
                sink.complete();
            } else {
                // 生产内容
                Util.print("生产者下发内容：" + state);
                sink.next(state);
                Util.sleep(100);
            }
            return ++state;
        }).publishOn(Schedulers.parallel(), 2).map(state -> {
            Util.print("将上游过来的数据state取反: " + -state);
            return -state;
        }).subscribe(
            // onNext
            (i) -> Util.print("订阅者接收到内容：" + i),
            // onError
            (e) -> e.printStackTrace(),
            // onComplete
            () -> {
                Util.print("Subscriber consume finished.");
                countDownLatch.countDown();
            });
        countDownLatch.await();
    }
}
